package com.cn.client.consumer;

import com.cn.client.MQClientInstance;
import com.cn.client.OffsetStore;
import com.cn.common.MessageQueue;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;

public class RemoteClientOffsetStore implements OffsetStore {

    private MQClientInstance mqClientInstance;

    /**
     * 存储偏移量
     */
    private ConcurrentMap<MessageQueue, AtomicLong> offsetTables = new ConcurrentHashMap<>();


    public RemoteClientOffsetStore(MQClientInstance mqClientInstance) {
        this.mqClientInstance = mqClientInstance;
    }


    /**
     * 修改本地消费位移
     *
     * @param messageQueue
     * @param offset
     */
    @Override
    public void updateOffset(MessageQueue messageQueue, Long offset) {
        if (offset != null) {
            AtomicLong old = this.offsetTables.get(messageQueue);
            if (old == null) {
                old = new AtomicLong(offset);
                old = this.offsetTables.putIfAbsent(messageQueue, old);
            }
            if (old != null) {
                old.set(offset);
            }
        }
    }

    /**
     * 从broker端查询消费位移
     *
     * @param messageQueue
     */
    public long readOffset(MessageQueue messageQueue) {
        long offset = mqClientInstance.getMqClientApi().getQueueOffset(messageQueue);
        this.offsetTables.putIfAbsent(messageQueue, new AtomicLong(offset));
        return offset;
    }

    /**
     * 持久化
     *
     * @param messageQueue
     */
    public void persit(MessageQueue messageQueue) {
        AtomicLong offset = this.offsetTables.get(messageQueue);
        mqClientInstance.getMqClientApi().persitQueueOffset(messageQueue, offset.get());
    }


    public void persitAll() {
        for (Map.Entry<MessageQueue, AtomicLong> map : this.offsetTables.entrySet()) {
            persit(map.getKey());
        }
    }

    @Override
    public void remove(MessageQueue mq) {
        this.offsetTables.remove(mq);
    }
}
